package com.cet.reactor;

import com.cet.reactor.exception.BusinessException;
import org.reactivestreams.Subscription;
import reactor.core.Disposable;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

import java.io.IOException;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * Flux 0 ~ N
 */
public class FluxDemo {

    public static void main(String[] args) throws IOException {

        Flux<Integer> just = Flux.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).delayElements(Duration.ofSeconds(1));

        just.subscribe(item -> System.out.println(Thread.currentThread().getName() + ":" + item));

        // can subscribe by multi consumers

        just.subscribe(item -> System.out.println(Thread.currentThread().getName() + ": consumer get :" + item));

        System.out.println("=======================");

        Flux<Object> emptyFlux = Flux.empty().doOnComplete(() -> System.out.println("stream completed"));

        emptyFlux.subscribe();


        Flux<Integer> rangeFlux = Flux.range(1, 10)
                .delayElements(Duration.ofSeconds(1))
                .doOnComplete(() -> System.out.println("stream completed"))
                .doOnCancel(() -> System.out.println("stream cancelled"))
                .doOnError(throwable -> System.out.println("stream error:" + throwable.getMessage()))
                .doOnNext(integer -> System.out.println("stream next:" + integer));

        Flux<Integer> filteredFlux = rangeFlux.filter(item -> item > 2);

        rangeFlux.subscribe(item -> System.out.println(Thread.currentThread().getName() + ":" + item));

        filteredFlux.subscribe(new BaseSubscriber<Integer>() {
            @Override
            protected Subscription upstream() {
                return super.upstream();
            }

            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                System.out.println("bind success");
                subscription.request(1);
            }

            @Override
            protected void hookOnNext(Integer value) {
                System.out.println("accept value:" + value);
                if (value <= 5) {
                    request(1);
                    Flux.just(1, 2, 0)
                            .map(i -> 1 / i)
                            .onErrorMap(original ->
                                    new BusinessException("oop, SLA exceeded", original)
                            )
                            .log()
                            .subscribe();
                } else {
                    cancel();
                }

            }

            @Override
            protected void hookOnComplete() {
                System.out.println("complete success");
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("error:" + throwable.getMessage());
            }

            @Override
            protected void hookOnCancel() {
                System.out.println("cancel success");
            }

            @Override
            protected void hookFinally(SignalType type) {
                // try finally
                super.hookFinally(type);
            }
        });

//        Flux<Long> intervalFlux = Flux.interval(Duration.ofSeconds(1));
//
//        intervalFlux.subscribe(item -> System.out.println(Thread.currentThread().getName() + ":" + item));

        System.in.read();
    }
}
